package com.zhenjun.dimension;

import com.alibaba.fastjson.JSONObject;
import com.zhenjun.domin.Order;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.cache.Cache;
import org.apache.flink.shaded.guava18.com.google.common.cache.CacheBuilder;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

/**
 * @author wangzj
 * @description LRU是一种缓存算法，意思是最近最少使用的数据则被淘汰。
 * @date 2020/8/18 21:55
 */
public class LRU extends RichAsyncFunction<String, Order> {
    private static final Logger log = LoggerFactory.getLogger(LRU.class);
    Cache<String, String> cache = null;
    private Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //创建hbase客户端
        connection = DriverManager.getConnection("jdbc:mysql://47.104.87.133:3306/dim?characterEncoding=UTF-8", "wangzj", "123456");
        cache = CacheBuilder.newBuilder()
                //最多存储10000条
                .maximumSize(10000)
                //过期时间为1分钟
                .expireAfterWrite(60, TimeUnit.SECONDS)
                .build();
    }

    @Override
    public void asyncInvoke(String input, ResultFuture<Order> resultFuture) throws SQLException {
        JSONObject jsonObject = JSONObject.parseObject(input);
        Integer cityId = jsonObject.getInteger("city_id");
        String userName = jsonObject.getString("user_name");
        String items = jsonObject.getString("items");
        //读缓存
        String cacheCityName = cache.getIfPresent(cityId);
        //如果缓存获取失败再从Mysql获取维度数据
        if (cacheCityName != null) {
            Order order = new Order();
            order.setCityId(cityId);
            order.setItems(items);
            order.setUserName(userName);
            order.setCityName(cacheCityName);
            resultFuture.complete(Collections.singleton(order));
        } else {
            String sql = "select city_name from city_info where city_id = ?";
            PreparedStatement ps = connection.prepareStatement(sql);
            //填充查询条件
            ps.setInt(1, cityId);
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                Order order = new Order();
                String cityName = rs.getString("city_name");
                order.setCityId(cityId);
                order.setCityName(cityName);
                order.setUserName(userName);
                order.setItems(items);
                resultFuture.complete(Collections.singleton(order));
                cache.put(String.valueOf(cityId), cityName);
            }
        }
    }
}
